package com.ruyuan.ingestion.kafka.streaming;
import com.ruyuan.ingestion.kafka.BaseStreamer;
import java.util.ArrayList;
import java.util.List;
public class UserFriendStreamer extends BaseStreamer {

    @Override
    protected String getApplicationId() {
        return "userFriend" + getTargetTopic();
    }

    @Override
    protected String getSourceTopic() {
        return "user_friends_raw";
    }

    @Override
    protected String getTargetTopic() {
        return "user_friends";
    }

    /**
     * 具体数据变换的方法
     *
     * fields数据样例：
     * 3197468391,1346449342 3873244116 4226080662 1222907620 547730952 1052032722 2138119761 417295859 1872292079 984265443 2535686531 3703382700 3581879482 2279455658 340978873 156124122 1470725855 4141612054 2696659498 2502277384 1741997451 2183113360 406756357 2170357047 1886083250 3160228214 818630635 3835934628
     * 3197468391，1346449342
     * 3197468391，3873244116
     * 3197468391，4226080662
     * ....
     * */
    @Override
    protected List<String[]> transform(String[] fields) {
        List<String[]> result = new ArrayList<>();
        //获取userId  -> 3197468391
        String userId = fields[0];
        String[] friends = fields[1].split(" ");
        if (fields != null && friends.length > 0) {
            for (String friend:fields) {
                result.add(new String[]{userId,friend});
            }
        }
        return result;
    }

    @Override
    protected Boolean isHeader(String[] fields) {
        return (isValid(fields) && fields[0].equals("user") && fields[1].equals("friends"));
    }

    @Override
    protected Boolean isValid(String[] fields) {
        return fields.length > 1;
    }
}
